package git.soulbgm.job;

import com.alibaba.druid.support.json.JSONUtils;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import git.soulbgm.common.annotation.ElasticDataSource;
import git.soulbgm.common.annotation.ElasticJob;
import git.soulbgm.common.enums.JobConfigMode;
import git.soulbgm.common.listener.ElasticJobListener;
import git.soulbgm.config.JobInfoConfig;
import git.soulbgm.mapper.JobInfoMapper;
import git.soulbgm.pojo.JobConfig;
import git.soulbgm.pojo.JobInfo;
import git.soulbgm.utils.JsonUtil;
import lombok.SneakyThrows;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.zookeeper.data.Stat;
import org.quartz.Job;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.CHILD_ADDED;

/**
 * 定时任务业务逻辑
 *
 * @author SoulBGM
 * @date 2020/11/18
 */
@Component
public class JobService {

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private ZookeeperRegistryCenter regCenter;

    @Autowired
    private JobInfoConfig jobConfig;

    @Resource
    private JobInfoMapper jobInfoMapper;

    private DataSource dataSource;

    @Bean
    public JobService initJob(DataSource dataSource, @ElasticDataSource ObjectProvider<DataSource> elasticDataSource) {
        this.dataSource = getDataSource(dataSource, elasticDataSource);
        JobConfigMode configMode = jobConfig.getConfigMode();
        if (configMode == JobConfigMode.ANNOTATION) {
            // 注解生成定时任务方式
            annotationInitJob();
        } else if (configMode == JobConfigMode.PROPERTIES) {
            // 配置文件生成定时任务方式
            propertiesInitJob();
        } else if (configMode == JobConfigMode.DATABASE) {
            // 数据库生成定时任务方式
            databaseInitJob();
        }
        monitorJobRegister();
        return this;
    }

    /**
     * 开启任务监听，当有任务添加时，监听zk中的数据添加，自动在其他节点也初始化该任务
     */
    public void monitorJobRegister() {
        CuratorFramework client = regCenter.getClient();
        @SuppressWarnings("resource")
        PathChildrenCache childrenCache = new PathChildrenCache(client, "/", true);
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                    Stat stat = client.checkExists().forPath(data.getPath() + "/config");
                    if (stat != null) {
                        String configJson = new String(client.getData().forPath(data.getPath() + "/config"));
                        JobConfig jobConfig = JsonUtil.fromJson(configJson, JobConfig.class);
                        JobInfo jobInfo = new JobInfo();
                        jobInfo.setJobBeanName(jobConfig.getJobClass());
                        jobInfo.setJobName(jobConfig.getJobName());
                        jobInfo.setCron(jobConfig.getCron());
                        jobInfo.setShardingItemParameters(jobConfig.getShardingItemParameters());
                        jobInfo.setShardingTotalCount(jobConfig.getShardingTotalCount());
                        jobInfo.setJobParameter(jobConfig.getJobParameter());
                        jobInfo.setDescription(jobConfig.getDescription());
                        initJob(jobInfo);
                    }
                }
            }
        };
        childrenCache.getListenable().addListener(childrenCacheListener);
        try {
            childrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private DataSource getDataSource(DataSource dataSource, ObjectProvider<DataSource> quartzDataSource) {
        DataSource dataSourceIfAvailable = quartzDataSource.getIfAvailable();
        return (dataSourceIfAvailable != null) ? dataSourceIfAvailable : dataSource;
    }

    /**
     * 数据库生成定时任务
     */
    private void databaseInitJob() {
        JobInfo jobInfo = new JobInfo();
        jobInfo.setStatus(1);
        List<JobInfo> infos = jobInfoMapper.select(jobInfo);
        initJob(infos);
    }

    /**
     * 配置文件生成定时任务
     */
    private void propertiesInitJob() {
        List<JobInfo> infos = jobConfig.getInfos();
        initJob(infos);
    }

    /**
     * 初始化定时任务
     *
     * @param infos 任务信息
     */
    private void initJob(List<JobInfo> infos) {
        for (JobInfo info : infos) {
            initJob(info);
        }
    }

    /**
     * 注解生成定时任务
     */
    private void annotationInitJob() {
        Map<String, SimpleJob> map = applicationContext.getBeansOfType(SimpleJob.class);
        List<JobInfo> infos = new ArrayList<>(map.size());
        for (Map.Entry<String, SimpleJob> entry : map.entrySet()) {
            SimpleJob simpleJob = entry.getValue();
            ElasticJob elasticSimpleJob = simpleJob.getClass().getAnnotation(ElasticJob.class);
            JobInfo info = new JobInfo();
            info.setJobBeanName(entry.getKey());
            info.setJobName(elasticSimpleJob.jobName());
            info.setCron(elasticSimpleJob.cron());
            info.setShardingItemParameters(elasticSimpleJob.shardingItemParameters());
            info.setShardingTotalCount(elasticSimpleJob.shardingTotalCount());
            info.setJobParameter(elasticSimpleJob.jobParameter());
            info.setJobEventRdb(elasticSimpleJob.jobEventRdb() ? 1 : 0);
            info.setJobListenerBeanName(elasticSimpleJob.jobListener());
            info.setDescription(elasticSimpleJob.description());
            infos.add(info);
        }
        initJob(infos);
    }

    /**
     * 初始化定时任务
     *
     * @param info      任务信息
     */
    @SneakyThrows
    public void initJob(JobInfo info) {
        String jobBeanName = info.getJobBeanName();
        Object bean = null;
        if (jobBeanName.contains(".")) {
            Class<?> jobBeanClass = Class.forName(jobBeanName);
            bean = applicationContext.getBean(jobBeanClass);
        } else {
            bean = applicationContext.getBean(jobBeanName);
        }
        if (bean instanceof SimpleJob) {
            SimpleJob simpleJob = (SimpleJob) bean;
            String cron = info.getCron();
            String jobName = StringUtils.defaultIfBlank(info.getJobName(), simpleJob.getClass().getName());
            SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(
                    JobCoreConfiguration
                            .newBuilder(jobName, cron, info.getShardingTotalCount())
                            .shardingItemParameters(info.getShardingItemParameters())
                            .jobParameter(info.getJobParameter()).build(),
                    simpleJob.getClass().getCanonicalName());
            LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration)
                    .overwrite(true).build();
            SpringJobScheduler jobScheduler = null;
            JobEventRdbConfiguration jobEventRdbConfiguration = null;
            ElasticJobListener elasticJobListener = null;
            if (info.getJobEventRdb() != null && info.getJobEventRdb() == 1 && dataSource != null) {
                jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource);
            }
            if (info.getJobListenerBeanName() != null && !"".equals(info.getJobListenerBeanName())) {
                elasticJobListener = (ElasticJobListener) applicationContext.getBean(info.getJobListenerBeanName());
            }
            if (jobEventRdbConfiguration != null && elasticJobListener != null) {
                jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration, elasticJobListener);
            } else if (jobEventRdbConfiguration != null) {
                jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration);
            } else {
                jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration);
            }
            jobScheduler.init();
        }
    }

}
